/*
 * Copyright 2021 Shulie Technology, Co.Ltd
 * Email: shulie@shulie.io
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.shulie.surge.data.runtime.disruptor;

import io.shulie.surge.data.runtime.disruptor.util.Util;

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

/**
 * Base class for the various sequencer types (single/multi).  Provides
 * common functionality like the management of gating sequences (add/remove) and
 * ownership of the current cursor.
 */
public abstract class AbstractSequencer implements Sequencer
{
	private static final AtomicReferenceFieldUpdater<AbstractSequencer, Sequence[]> SEQUENCE_UPDATER =
			AtomicReferenceFieldUpdater.newUpdater(AbstractSequencer.class, Sequence[].class, "gatingSequences");

	protected final int bufferSize;
	protected final WaitStrategy waitStrategy;
	protected final Sequence cursor = new Sequence(INITIAL_CURSOR_VALUE);
	protected volatile Sequence[] gatingSequences = new Sequence[0];

	/**
	 * Create with the specified buffer size and wait strategy.
	 *
	 * @param bufferSize The total number of entries, must be a positive power of 2.
	 * @param waitStrategy
	 */
	public AbstractSequencer(int bufferSize, WaitStrategy waitStrategy)
	{
		if (bufferSize < 1)
		{
			throw new IllegalArgumentException("bufferSize must not be less than 1");
		}
		if (Integer.bitCount(bufferSize) != 1)
		{
			throw new IllegalArgumentException("bufferSize must be a power of 2");
		}

		this.bufferSize = bufferSize;
		this.waitStrategy = waitStrategy;
	}

	/**
	 * @see Sequencer#getCursor()
	 */
	@Override
	public final long getCursor()
	{
		return cursor.get();
	}

	/**
	 * @see Sequencer#getBufferSize()
	 */
	@Override
	public final int getBufferSize()
	{
		return bufferSize;
	}

	/**
	 * @see Sequencer#addGatingSequences(Sequence...)
	 */
	@Override
	public final void addGatingSequences(Sequence... gatingSequences)
	{
		SequenceGroups.addSequences(this, SEQUENCE_UPDATER, this, gatingSequences);
	}

	/**
	 * @see Sequencer#removeGatingSequence(Sequence)
	 */
	@Override
	public boolean removeGatingSequence(Sequence sequence)
	{
		return SequenceGroups.removeSequence(this, SEQUENCE_UPDATER, sequence);
	}

	/**
	 * @see Sequencer#getMinimumSequence()
	 */
	@Override
	public long getMinimumSequence()
	{
		return Util.getMinimumSequence(gatingSequences, cursor.get());
	}

	/**
	 * @see Sequencer#newBarrier(Sequence...)
	 */
	@Override
	public SequenceBarrier newBarrier(Sequence... sequencesToTrack)
	{
		return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
	}
}